package task3

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/*import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.util.Collector*/

import java.text.SimpleDateFormat

/**
 * @author jhhe66
 * @date 2021/6/18 10:11
 */
object TumblingWindowDemo {
  def main(args: Array[String]): Unit = {
    // 获取流环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 设置并行度
    env.setParallelism(1)

    // 自定义数据源生成数据，每秒生成一条当前时间戳
    val dataStream: KeyedStream[(String, Long), String] = env.addSource(new SourceFunction[(String, Long)] {
      var flag = true

      override def run(sourceContext: SourceFunction.SourceContext[(String, Long)]): Unit = {
        while (flag) {
          sourceContext.collect(("hello", System.currentTimeMillis()))
          Thread.sleep(1000)
        }
      }

      override def cancel(): Unit = {
        flag = false
      }
    }).keyBy(_._1)

    // 将key流转为窗口流，设置窗口长度为10秒
    val windowStream: WindowedStream[(String, Long), String, TimeWindow] = dataStream.timeWindow(Time.seconds(10))

    val res = windowStream.apply(new WindowFunction[(String, Long), (String, Long), String, TimeWindow] {
      override def apply(key: String, window: TimeWindow, input: Iterable[(String, Long)], out: Collector[(String, Long)]): Unit = {
        val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
        input.foreach(tup => out.collect((s"${tup._1},开始时间:${sdf.format(window.getStart)},结束时间:${sdf.format(window.getEnd)},value:${sdf.format(tup._2)}", tup._2)))
        println("=================================================本次窗口结束====================================================================")
      }
    })

    res.print()
    env.execute()
  }
}
